package com.zyb.dfs.namenode.server;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zyb.dfs.namenode.rpc.model.*;
import com.zyb.dfs.namenode.rpc.service.NameNodeServiceGrpc;
import io.grpc.stub.StreamObserver;

import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;

/**
 * NameNode的rpc服务的接口
 *
 * @author zhonghuashishan
 */
public class NameNodeServiceImpl extends NameNodeServiceGrpc.NameNodeServiceImplBase {

    public static final Integer STATUS_SUCCESS = 1;
    public static final Integer STATUS_FAILURE = 2;
    public static final Integer STATUS_SHUTDOWN = 3;
    public static final Integer STATUS_DUPLICATE = 4;

    public static final Integer BACKUP_NODE_FETCH_SIZE = 10;

    /**
     * 负责管理元数据的核心组件
     */
    private FSNamesystem namesystem;

    /**
     * 是否还在运行
     */
    private volatile Boolean isRunning = true;

    /**
     * 负责管理集群中所有datanode的组件
     */
    private DataNodeManager dataNodeManager;


    /**
     * 当前缓冲的一小部分editslog
     */
    private JSONArray currentBufferEditsLog = new JSONArray();

    /**
     * 当前内存里缓冲了哪个磁盘文件的数据
     */
    private String bufferedFlushedTxid;

    /**
     * 当前内存缓冲区最大的txid
     */
    private long currentBufferdMaxTxid = 0L;

    public NameNodeServiceImpl(
            FSNamesystem namesystem,
            DataNodeManager dataNodeManager) {
        this.namesystem = namesystem;
        this.dataNodeManager = dataNodeManager;
    }

    /**
     * datanode进行注册
     *
     * @return
     * @throws Exception
     */
    @Override
    public void register(RegisterRequest request, StreamObserver<RegisterResponse> responseObserver) {
        Boolean result = this.dataNodeManager.register(request.getIp(), request.getHostname(), request.getNioPort());
        RegisterResponse response = null;
        if (result) {
            response = RegisterResponse.newBuilder().setStatus(STATUS_SUCCESS).build();
        } else {
            response = RegisterResponse.newBuilder().setStatus(STATUS_FAILURE).build();
        }
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void heartbeat(HeartbeatRequest request, StreamObserver<HeartbeatResponse> responseObserver) {
        String ip = request.getIp();
        String hostname = request.getHostname();
        Boolean result = this.dataNodeManager.heartbeat(request.getIp(), request.getHostname());
        HeartbeatResponse response = null;
        List<Command> commands = new ArrayList<>();
        if (result) {
            // 如果心跳成功了，此时应该查看一下是否有复制副本的任务
            // 如果有，则做成命令下发给这个数据节点
            DataNodeInfo datanode = dataNodeManager.getDataNodeInfo(ip, hostname);
            ReplicateTask replicateTask = null;
            while ((replicateTask = datanode.pollReplicateTask()) != null) {
                Command replicateCommand = new Command(Command.REPLICATE);
                replicateCommand.setContent(JSONObject.toJSONString(replicateTask));
                commands.add(replicateCommand);
            }

            RemoveReplicaTask removeReplicaTask = null;
            while ((removeReplicaTask = datanode.pollRemoveReplicaTask()) != null) {
                Command removeReplicaCommand = new Command(Command.REMOVE_REPLICA);
                removeReplicaCommand.setContent(JSONObject.toJSONString(removeReplicaTask));
                commands.add(removeReplicaCommand);
            }

            response = HeartbeatResponse.newBuilder()
                    .setStatus(STATUS_SUCCESS)
                    .setCommands(JSONArray.toJSONString(commands))
                    .build();
        } else {
            Command registerCommand = new Command(Command.REGISTER);
            Command reportCompleteStorageInfoCommand = new Command(
                    Command.REPORT_COMPLETE_STORAGE_INFO);
            commands.add(registerCommand);
            commands.add(reportCompleteStorageInfoCommand);

            response = HeartbeatResponse.newBuilder()
                    .setStatus(STATUS_FAILURE)
                    .setCommands(JSONArray.toJSONString(commands))
                    .build();
        }

        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 创建目录
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void mkdir(MkdirRequest request, StreamObserver<MkdirResponse> responseObserver) {
        try {
            MkdirResponse response = null;
            if (!isRunning) {
                response = MkdirResponse.newBuilder().setStatus(STATUS_SHUTDOWN).build();
            } else {
                this.namesystem.mkdir(request.getPath());
                response = MkdirResponse.newBuilder().setStatus(STATUS_SUCCESS).build();
            }
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            e.getStackTrace();
        }
    }

    /**
     * 优雅关闭的接口
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void shutdown(ShutdownRequest request, StreamObserver<ShutdownResponse> responseObserver) {
        this.isRunning = false;
        this.namesystem.flush();
        this.namesystem.saveCheckpointTxid();
        ShutdownResponse response = ShutdownResponse.newBuilder().setStatus(STATUS_SHUTDOWN).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 拉取editslog接口
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void fetchEditsLog(FetchEditsLogRequest request, StreamObserver<FetchEditsLogResponse> responseObserver) {
        if (!isRunning) {
            FetchEditsLogResponse response = FetchEditsLogResponse.newBuilder()
                    .setEditsLog(new JSONArray().toJSONString())
                    .build();
            responseObserver.onNext(response);
            responseObserver.onCompleted();
            return;
        }
        long syncTxid = request.getSyncedTxid();

        FetchEditsLogResponse response = null;
        JSONArray fetchedEditsLog = new JSONArray();

        List<String> flushedTxids = namesystem.getEditlog().getFlushedTxids();

        // 如果此时还没有刷出来任何磁盘文件的话，那么此时数据仅仅存在于内存缓冲里
        if (flushedTxids.size() == 0) {
            fetchFromBufferedEditsLog(syncTxid, fetchedEditsLog);
        }
        // 如果此时发现已经有落地磁盘的文件了，这个时候就要扫描所有的磁盘文件的索引范围
        else {
            // 第一种情况，你要拉取的txid是在某个磁盘文件里的
            // 有磁盘文件，而且内存里还缓存了某个磁盘文件的数据了
            if (bufferedFlushedTxid != null) {
                // 如果要拉取的数据就在当前缓存的磁盘文件数据里
                if (existInFlushedFile(syncTxid, bufferedFlushedTxid)) {
                    fetchFromCurrentBuffer(syncTxid, fetchedEditsLog);
                }
                // 如果要拉取的数据不在当前缓存的磁盘文件数据里了，那么需要从下一个磁盘文件去拉取
                else {
                    String nextFlushedTxid = getNextFlushedTxid(flushedTxids, bufferedFlushedTxid);
                    // 如果可以找到下一个磁盘文件，那么就从下一个磁盘文件里开始读取数据
                    if (nextFlushedTxid != null) {
                        fetchFromFlushedFile(syncTxid, nextFlushedTxid, fetchedEditsLog);
                    }
                    // 如果没有找到下一个文件，此时就需要从内存里去继续读取
                    else {
                        fetchFromBufferedEditsLog(syncTxid, fetchedEditsLog);
                    }
                }
            } else {
                // 遍历所有的磁盘文件的索引范围，0-390，391-782
                Boolean fechedFromFlushedFile = false;

                for (String flushedTxid : flushedTxids) {
                    // 如果要拉取的下一条数据就是在某个磁盘文件里
                    if (existInFlushedFile(syncTxid, flushedTxid)) {
                        // 此时可以把这个磁盘文件里以及下一个磁盘文件的的数据都读取出来，放到内存里来缓存
                        // 就怕一个磁盘文件的数据不足够10条
                        fetchFromFlushedFile(syncTxid, flushedTxid, fetchedEditsLog);
                        fechedFromFlushedFile = true;
                        break;
                    }
                }

                // 第二种情况，你要拉取的txid已经比磁盘文件里的全部都新了，还在内存缓冲里
                // 如果没有找到下一个文件，此时就需要从内存里去继续读取
                if (!fechedFromFlushedFile) {
                    fetchFromBufferedEditsLog(syncTxid, fetchedEditsLog);
                }
            }
        }

        response = FetchEditsLogResponse.newBuilder()
                .setEditsLog(fetchedEditsLog.toJSONString()) // []
                .build();

        responseObserver.onNext(response);
        responseObserver.onCompleted();

    }

    /**
     * 获取下一个磁盘文件对应的txid范围
     *
     * @param flushedTxids
     * @param bufferedFlushedTxid
     * @return
     */
    private String getNextFlushedTxid(List<String> flushedTxids, String bufferedFlushedTxid) {
        for (int i = 0; i < flushedTxids.size(); i++) {
            if (flushedTxids.get(i).equals(bufferedFlushedTxid)) {
                if (i + 1 < flushedTxids.size()) {
                    return flushedTxids.get(i + 1);
                }
            }
        }
        return null;
    }

    /**
     * 从已经刷入磁盘的文件里读取editslog，同时缓存这个文件数据到内存
     *
     * @param flushedTxid
     */
    private void fetchFromFlushedFile(long syncTxid, String flushedTxid, JSONArray fetchedEditsLog) {
        try {
            String[] flushedTxidSplited = flushedTxid.split("_");
            long startTxid = Long.valueOf(flushedTxidSplited[0]);
            long endTxid = Long.valueOf(flushedTxidSplited[1]);

            String currentEditsLogFile = "/Users/admin/editslog/edits-"
                    + startTxid + "-" + endTxid + ".log";
            List<String> editsLogs = Files.readAllLines(Paths.get(currentEditsLogFile),
                    StandardCharsets.UTF_8);

            currentBufferEditsLog.clear();
            for (String editsLog : editsLogs) {
                currentBufferEditsLog.add(JSONObject.parseObject(editsLog));
                currentBufferdMaxTxid = JSONObject.parseObject(editsLog).getLong("TXID");
            }
            bufferedFlushedTxid = flushedTxid; // 缓存了某个刷入磁盘文件的数据

            fetchFromCurrentBuffer(syncTxid, fetchedEditsLog);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 是否存在于刷到磁盘的文件中
     *
     * @param flushedTxid
     * @return
     */
    private Boolean existInFlushedFile(long syncTxid, String flushedTxid) {
        String[] flushedTxidSplited = flushedTxid.split("_");

        long startTxid = Long.valueOf(flushedTxidSplited[0]);
        long endTxid = Long.valueOf(flushedTxidSplited[1]);
        long fetchTxid = syncTxid + 1;

        if (fetchTxid >= startTxid && fetchTxid <= endTxid) {
            return true;
        }

        return false;
    }

    /**
     * 就是从内存缓冲的editslog中拉取数据
     *
     * @param fetchedEditsLog
     */
    private void fetchFromBufferedEditsLog(long syncTxid, JSONArray fetchedEditsLog) {
        // 如果要拉去的txid还在上一次的内存缓存中
        long fetchTxid = syncTxid + 1;
        if (fetchTxid <= currentBufferdMaxTxid) {
            fetchFromCurrentBuffer(syncTxid, fetchedEditsLog);
            return;
        }

        currentBufferEditsLog.clear();

        String[] bufferedEditsLog = namesystem.getEditlog().getBufferedEditsLog();
        if (bufferedEditsLog != null) {
            for (String editsLog : bufferedEditsLog) {
                currentBufferEditsLog.add(JSONObject.parseObject(editsLog));
                // 我们在这里要记录下内存缓存最大的txid是多少，下一次再拉取走下判断，之前内存缓存中数据是否话可以读取
                // 不要每次都从内存缓冲中加载
                currentBufferdMaxTxid = JSONObject.parseObject(editsLog).getLong("TXID");
            }

            bufferedFlushedTxid = null;
        }
    }

    /**
     * 从当前已经在内存里缓存的数据中拉取editslog
     *
     * @param fetchedEditsLog
     */
    private void fetchFromCurrentBuffer(long syncTxid, JSONArray fetchedEditsLog) {
        int fetchCount = 0;
        for (int i = 0; i < currentBufferEditsLog.size(); i++) {
            if (currentBufferEditsLog.getJSONObject(i).getLong("TXID") == syncTxid + 1) {
                fetchedEditsLog.add(currentBufferEditsLog.getJSONObject(i));
                syncTxid = currentBufferEditsLog.getJSONObject(i).getLong("TXID");
                fetchCount++;
            }
            if (fetchCount == BACKUP_NODE_FETCH_SIZE) {
                break;
            }
        }
    }

    /**
     * 更新checkpoint的txid
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void updateCheckpointTxid(UpdateCheckpointTxidRequest request, StreamObserver<UpdateCheckpointTxidResponse> responseObserver) {
        long txid = request.getTxid();
        namesystem.setCheckpointTxid(txid);
        UpdateCheckpointTxidResponse response = UpdateCheckpointTxidResponse.newBuilder()
                .setStatus(1)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 创建文件
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void createFile(CreateFileRequest request, StreamObserver<CreateFileResponse> responseObserver) {
        try {
            CreateFileResponse response = null;

            if (!isRunning) {
                response = CreateFileResponse.newBuilder()
                        .setStatus(STATUS_SHUTDOWN)
                        .build();
            } else {
                String filename = request.getFilename();
                Boolean success = namesystem.create(filename);

                if (success) {
                    response = CreateFileResponse.newBuilder()
                            .setStatus(STATUS_SUCCESS)
                            .build();
                } else {
                    response = CreateFileResponse.newBuilder()
                            .setStatus(STATUS_DUPLICATE)
                            .build();
                }
            }

            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 为文件上传分配datanode节点
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void allocateDataNodes(AllocateDataNodesRequest request, StreamObserver<AllocateDataNodesResponse> responseObserver) {
        // 请出来所有的datanode，并且按照存储数据大小来排序，选择最少的两个出来

        // 默认认为要上传的文件会被放到那两个DataNode上去，此时我们就应该更新那两个DataNode存储数据的大小
        // 只有这样，后面的上传请求才能基于最新的存储情况

        long filesize = request.getFileSize();
        List<DataNodeInfo> dataNodeInfos = dataNodeManager.allocateDataNodes(filesize);

        AllocateDataNodesResponse response = AllocateDataNodesResponse.newBuilder()
                .setDatanodes(JSONUtil.toJsonStr(dataNodeInfos))
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 数据节点通知自己接受到副本
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void informReplicaReceived(InformReplicaReceivedRequest request, StreamObserver<InformReplicaReceivedResponse> responseObserver) {
        String hostname = request.getHostname();
        String filename = request.getFilename();
        String ip = request.getIp();

        try {
            namesystem.addReceiveReplica(hostname, ip, filename, 10);
        } catch (Exception e) {
        }
        InformReplicaReceivedResponse response = InformReplicaReceivedResponse.newBuilder().setStatus(1).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 上报全量信息
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void reportCompleteStorageInfo(ReportCompleteStorageInfoRequest request, StreamObserver<ReportCompleteStorageInfoResponse> responseObserver) {
        String hostname = request.getHostname();
        String filenames = request.getFilenames();
        String ip = request.getIp();
        Long storedDataSize = request.getStoredDataSize();
        dataNodeManager.setStoredDataSize(ip, hostname, storedDataSize);
        JSONArray filenamesObj = JSONArray.parseArray(filenames);
        for (int i = 0; i < filenamesObj.size(); i++) {
            String filename = filenamesObj.getString(i);
            namesystem.addReceiveReplica(hostname, ip, filename, storedDataSize);
        }

        ReportCompleteStorageInfoResponse response = ReportCompleteStorageInfoResponse.newBuilder().setStatus(1).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 获取文件副本所在datanode
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void getDataNodeForFile(GetDataNodeForFileRequest request, StreamObserver<GetDataNodeForFileResponse> responseObserver) {
        String filename = request.getFilename();
        String excludedDataNodeId = request.getExcludedDataNodeId();
        DataNodeInfo dataNodeInfo = namesystem.getDataNodeForFile(filename,excludedDataNodeId);
        GetDataNodeForFileResponse response = GetDataNodeForFileResponse.newBuilder().setDatanodeInfo(JSONArray.toJSONString(dataNodeInfo)).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 重新分配数据接口
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void reallocateDataNode(ReallocateDataNodeRequest request, StreamObserver<ReallocateDataNodeResponse> responseObserver) {
        long fileSize = request.getFileSize();
        String excludedDataNodeId = request.getExcludedDataNodeId();
        DataNodeInfo datanode = dataNodeManager.reallocateDataNode(fileSize, excludedDataNodeId);

        ReallocateDataNodeResponse response = ReallocateDataNodeResponse.newBuilder()
                .setDatanodeInfo(JSONObject.toJSONString(datanode))
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 集群重平衡
     * @param request
     * @param responseObserver
     */
    @Override
    public void rebalance(RebalanceRequest request, StreamObserver<RebalanceResponse> responseObserver) {

    }
}
